package com.yb.handler;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class MessageQueue {

	//高并发  互斥锁和条件变量
	Lock lock;
	
	Condition notEmpty;
	Condition notFull;
	
	
	//存贮消息列表
	Message[] items;
	
	int count;
	
	int takeIndex;
	
	int putIndex;
	
	public MessageQueue(){
		//消息要有个容量
		this.items = new Message[50];
		this.lock = new ReentrantLock();
		this.notEmpty = lock.newCondition();
		this.notFull = lock.newCondition();
		
	}
	/**
	 * 加入消息队列 (生产者，子线程，当消息队列已满，停止生产，子线程阻塞)
	 * @param msg
	 */
	public void enqueueMessage(Message msg){
		
		try {
			
			lock.lock();
			//判断多次 防止唤醒同类 所以把if改为While
			while (count == items.length) {
				//也可以用msg.wait()
				//阻塞
				try {
					notFull.await();
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
			items[putIndex] = msg;
			putIndex = (++putIndex == items.length) ? 0:putIndex;
			count++;
			notEmpty.signalAll();//通知消费者可以消费了
			
		} catch (Exception e) {
			// TODO: handle exception
		} finally{
			lock.unlock();
		}
		
	}
	
	/**
	 * 消息出队  (消费者,主线程，当消息队列被掏空，停止消费,主线程阻塞）
	 * @return
	 */
	public Message next(){
		Message msg = null;
		try {
			lock.lock();
			while ( count == 0) {
				notEmpty.await();
			}
			msg = items[takeIndex];
			takeIndex = (++takeIndex == items.length) ? 0:takeIndex;
			count--;
			notFull.signalAll();//通知生产这生产
		} catch (Exception e) {
			// TODO: handle exception
		} finally{
			lock.unlock();
		}
		
		return msg;
	}
}
